package org.example.window;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.example.data.WaterSensor;
import org.example.function.WaterSensorMapFunction;

public class WindowApiDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<WaterSensor> wsSource = source.map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedStream = wsSource.keyBy(WaterSensor::getId);
        //1、指定窗口分配器，用哪一种窗口
        //a.没有keyBy，窗口中所有数据都在同一个子任务处理，强制并行度为1.
        //source.windowAll()
        //2、指定KeyBy，每个key都定义了一组窗口，各自进行计算

        //基于时间的
        //每10秒滚动一次
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        //滑动窗口
        //WindowedStream<WaterSensor, String, TimeWindow> slidingWindow = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2)));
        //会话窗口
        //keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        //基于计数的
        //表示滚动计数窗口，窗口长度为5
        //keyedStream.countWindow(5);
        //表示滑动计数窗口，长度为5，步长为2
        //keyedStream.countWindow(5,2);

        //窗口函数
        //增量聚合，来一条算一条,窗口触发的时候，输出计算结果，有：reduce,aggregate

        //全窗口函数，数据来了不计算，存起来，窗口触发的时候，计算并输出结果，有：process


        env.execute("Window API DEMO");
    }
}
